package drds.binlog.prometheus.impl;

import com.google.common.base.Preconditions;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.instance.core.BinlogTask;
import drds.binlog.metadata.MetaDataManager;
import drds.binlog.prometheus.InstanceExports;
import drds.binlog.prometheus.InstanceRegistry;
import io.prometheus.client.Collector;
import io.prometheus.client.GaugeMetricFamily;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;


public class MetaCollector extends Collector implements InstanceRegistry
{

    private static final List<String> INFO_LABELS_LIST = Arrays.asList("destination", "mode");
    private static final Logger logger = LoggerFactory.getLogger(MetaCollector.class);
    private static final String INSTANCE = "canal_instance";
    private static final String INSTANCE_HELP = "Info instance";
    private static final String SUBSCRIPTION = "canal_instance_subscriptions";
    private static final String SUBSCRIPTION_HELP = "Info instance subscriptions";
    private final ConcurrentMap<String, MetaMetricsHolder> destinationToMetaMetricsHolderMap = new ConcurrentHashMap<String, MetaMetricsHolder>();

    private MetaCollector()
    {
    }

    public static MetaCollector instance()
    {
        return SingletonHolder.SINGLETON;
    }

    @Override
    public List<MetricFamilySamples> collect()
    {
        List<MetricFamilySamples> metricFamilySamplesList = new ArrayList<MetricFamilySamples>();
        GaugeMetricFamily gaugeMetricFamily = new GaugeMetricFamily(INSTANCE,
                INSTANCE_HELP, INFO_LABELS_LIST);
        GaugeMetricFamily subsInfo = new GaugeMetricFamily(SUBSCRIPTION,
                SUBSCRIPTION_HELP, InstanceExports.DEST_LABELS_LIST);
        for (Map.Entry<String, MetaMetricsHolder> entry : destinationToMetaMetricsHolderMap.entrySet())
        {
            final String destination = entry.getKey();
            final MetaMetricsHolder metricsHolder = entry.getValue();
            gaugeMetricFamily.addMetric(metricsHolder.infoLabelValues, 1);
            List<ClientInfo> clientInfoList = metricsHolder.metaDataManager.getClientInfoList(destination);
            int count = clientInfoList == null ? 0 : clientInfoList.size();
            subsInfo.addMetric(metricsHolder.destLabelValues, count);
        }
        metricFamilySamplesList.add(gaugeMetricFamily);
        metricFamilySamplesList.add(subsInfo);
        return metricFamilySamplesList;
    }

    @Override
    public void register(BinlogTask binlogTask)
    {
        final String destination = binlogTask.getDestination();
        MetaMetricsHolder metaMetricsHolder = new MetaMetricsHolder();
        metaMetricsHolder.infoLabelValues = Arrays.asList(destination, "");
        metaMetricsHolder.destLabelValues = Collections.singletonList(destination);
        metaMetricsHolder.metaDataManager = binlogTask.getMetaDataManager();
        Preconditions.checkNotNull(metaMetricsHolder.metaDataManager);
        MetaMetricsHolder old = destinationToMetaMetricsHolderMap.put(destination, metaMetricsHolder);
        if (old != null)
        {
            logger.warn("Remove stale MetaCollector for binlogTask {}.", destination);
        }
    }

    @Override
    public void unregister(BinlogTask binlogTask)
    {
        final String destination = binlogTask.getDestination();
        destinationToMetaMetricsHolderMap.remove(destination);
    }

    private static class SingletonHolder
    {
        private static final MetaCollector SINGLETON = new MetaCollector();
    }

    private class MetaMetricsHolder
    {
        private List<String> infoLabelValues;
        private List<String> destLabelValues;
        private MetaDataManager metaDataManager;
    }


}
